package com.zx.mes.hyl.rocket.mq.consume;

import com.zx.mes.hyl.rocket.mq.util.RocketUtil;
import com.zx.mes.hyl.rocket.mq.yml.RocketMqProperties;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * rocket mq transaction
 * 与平时普通的consumer没什么区别
 *
 * @author 华云龙
 * @date 2018-12-3
 */
@Log4j2
public class CommonConsumer {

    private String producerGroupName = "tx-long-order-add-producer-group-name";

    private String subExpression = "*";

    private RocketMqProperties rocketMqProperties;

    private IConsumeMessage consumeMessage;


    /**
     * 配置consumer (推荐使用此方法进行配置)
     *
     * @param producerGroupName  生产者组
     * @param rocketMqProperties rocket mq yml properties
     * @param consumeMessage     实现接口IConsumeMessage
     * @return this 此类的实例对象
     */
    public CommonConsumer consumerConfig(String producerGroupName, RocketMqProperties rocketMqProperties, IConsumeMessage consumeMessage) {
        this.producerGroupName = producerGroupName;
        this.rocketMqProperties = rocketMqProperties;
        this.consumeMessage = consumeMessage;
        return this;
    }

    /**
     * 配置consumer
     *
     * @param producerGroupName  生产者组
     * @param rocketMqProperties rocket mq yml properties
     * @param consumeMessage     实现接口IConsumeMessage
     * @param subExpression      过虑消息条件,一般不怎么使用
     * @return this 此类的实例对象
     */
    public CommonConsumer consumerConfig(String producerGroupName, RocketMqProperties rocketMqProperties, IConsumeMessage consumeMessage, String subExpression) {
        this.subExpression = subExpression;
        return this.consumerConfig(producerGroupName, rocketMqProperties, consumeMessage);
    }

    public void consumer() throws MQClientException {

        /// 第一步:检测rocket mq 配置是否准确
        RocketMqProperties.RocketMq rocketMq = RocketUtil.checkRocketMqConfig(this.producerGroupName, this.rocketMqProperties);

        if (this.consumeMessage == null) {
            log.error("请配置IConsumeMessage consumeMessage:[" + this.consumeMessage + "]");
            return;
        }
        if (rocketMq != null && StringUtils.isNotBlank(rocketMq.getProducerGroupName())) {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();

            /// 设置消费群组
            consumer.setConsumerGroup(rocketMq.getProducerGroupName());

            /// 消费线程配置
            consumer.setConsumeThreadMin(rocketMq.getConsumeThreadMin());
            consumer.setConsumeThreadMax(rocketMq.getConsumeThreadMax());

            /// rocket mq 服务器
            consumer.setNamesrvAddr(rocketMq.getNameServer());

            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

            /// 注册
            consumer.subscribe(rocketMq.getTopic(), this.subExpression);

            /// 监听
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                MessageExt messageExt = msgs.get(0);
                int times = messageExt.getReconsumeTimes();

                boolean b = this.consumeMessage.consumeMessage(msgs, context);
                //noinspection StatementWithEmptyBody
                if (b) {
                } else {
                    if (times >= rocketMq.getConsumeRetryTimes()) {
                        log.debug("当前失败尝试消费次数为:[" + times + "]次,记录失败的消息,进行人工补偿,失败的消息:[" + messageExt + "]");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            log.info("开始消费生产者组:[" + rocketMq.getProducerGroupName() + "中的内容!");
        } else {
            log.error("你的rocket mq配置信息错误,请检查你生产者组配置是否存在!");
        }
    }
}
